package com.sdlh.demo.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.IOUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.io.FileOutputStream;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class MqttExample {
    final static String broker = "tcp://1.95.170.86:1883";
    final static String clientId = "JavaSample";
    final static String userName = "admin";
    final static String password = "zzkj688737";
    final static String device_report_push_topic = "1/device/0014975e87b7/report/push";
    final static String device_file_request_topic = "1/device/0014975e87b7/file/request";

    public static void main(String[] args) {
        MemoryPersistence persistence = new MemoryPersistence();

        try {
            MqttClient client = new MqttClient(broker, clientId, persistence);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setCleanSession(true);
            connOpts.setUserName(userName);
            connOpts.setPassword(password.toCharArray());

            log.info("Connecting to broker: " + broker);
            client.connect(connOpts);
            log.info("Connected");

            AtomicInteger count = new  AtomicInteger();
            client.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable throwable) {
                    log.info("断开了MQTT连接 ：{}", throwable.getMessage());
                    log.error(throwable.getMessage(), throwable);
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    //  TODO    此处可以将订阅得到的消息进行业务处理、数据存储
                    log.info("收到来自主题[" + topic + "]的消息：{}",
                             new String(message.getPayload()));
                    try(FileOutputStream outputStream = new FileOutputStream("device_file_" + count.incrementAndGet() + ".txt")) {
                        IOUtils.write(message.getPayload(), outputStream);
                    } catch (Exception e) {
                        log.error(e.getMessage(), e);
                    }
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                    try {
                        log.info("-------发布消息成功:{}-------",
                                 new String(iMqttDeliveryToken.getMessage().getPayload()));
                    } catch (MqttException e) {
                        throw new RuntimeException(e);
                    }
                }
            });

            // 订阅主题
            client.subscribe(device_file_request_topic);

            // 发布消息
//            String content = "Hello MQTT";
//            int qos = 2;
//            MqttMessage message = new MqttMessage(content.getBytes());
//            message.setQos(qos);
//            client.publish(DEVICE_REPORT_TOPIC, message);

            // 断开连接
//            client.disconnect();
        } catch (Exception e) {
            log.error("msg:" + e.getMessage(), e);
        }
    }
}
